package com.jltx.rx;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

/**
 * @author jltxseo
 *         Created by junlintianxia on 2017/08/10.
 */

public class RxJava2Learn3 {

    public  static void main(String[] args){
        RxJava2Learn3 rxJava = new RxJava2Learn3();
        rxJava.rxJavaScheduler();
    }

    /**
     * 错误处理,所有的错误都集合一起
     */
    public void erroHanderFlowable(){
        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> e) throws Exception {
                e.onNext("exception:" + (1 / 0));
                e.onComplete();
            }
        }, BackpressureStrategy.BUFFER)
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        System.out.println("onSubscribe");
                    }

                    @Override
                    public void onNext(String s) {
                        System.out.println("onNext:"+s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        System.out.println("onError:"+t.toString());
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("onComplete");
                    }
                });

    }

    public void erroHanderFlowable1(){
        Flowable.just(1,2,3,4)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        System.out.println("accept.onNext:"+Thread.currentThread());
                        System.out.println("onNext:"+integer/0);
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        System.out.println("accept.onError:"+Thread.currentThread());
                        System.out.println("onError:"+throwable.toString());
                    }
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        System.out.println("accept.thread1:"+Thread.currentThread());
                        System.out.println("onComplete");
                    }
                });

    }

    public void rxJavaScheduler(){
        List<Integer> list = new ArrayList<>();
        list.add(10);
        list.add(1);
        list.add(20);

        Flowable.just(list)
                .subscribeOn(Schedulers.single())// 指定 subscribe() 发生在 IO 线程
                .flatMap(new Function<List<Integer>, Publisher<Integer>>() {
                    @Override
                    public Publisher<Integer> apply(List<Integer> integers) throws Exception {
                        System.out.println("flatMap.thread:"+Thread.currentThread());
                        return Flowable.fromIterable(integers);
                    }
                })
                .map(new Function<Integer, String>() {
                    @Override
                    public String apply(Integer integer) throws Exception {
                        System.out.println("map.thread:"+Thread.currentThread());
                        return integer.toString();
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println("subscribe.thread:"+Thread.currentThread());
                        System.out.println("accept:"+s);
                    }
                });
    }


    /**
     * RxJava线程控制Scheduler
     */
    public void rxJavaScheduler1(){
        Observable.just(1,2,3,4)
                .observeOn(Schedulers.trampoline())  //指定 Subscriber 的回调直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        System.out.println("accept.onNext:"+Thread.currentThread());
                        System.out.println("onNext:"+integer);
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        System.out.println("accept.thread1:"+Thread.currentThread());
                        System.out.println("onError:"+throwable.toString());
                    }
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        System.out.println("accept.thread1:"+Thread.currentThread());
                        System.out.println("onComplete");
                    }
                });
    }
}
